package com.example.controller;

import com.example.modle.ResponseInfo;
import com.example.modle.User;
import com.example.service.HttpService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult;
import rx.Observable;
import rx.schedulers.Schedulers;

import javax.servlet.http.HttpServletResponse;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicReference;


/**
 * DeferredResult 参考：https://blog.csdn.net/miaomiao19971215/article/details/106328609
 *
 * Servlet3.0提供了基于servlet的异步处理api，Spring MVC只是将这些api进行了一系列的封装，从而实现了DeferredResult。
 *
 * DeferredResult字面意思是"延迟结果"，它允许Spring MVC收到请求后，立即释放(归还)容器线程，以便容器可以接收更多的外部请求，提升吞吐量，与此同时，DeferredResult将陷入阻塞，直到我们主动将结果set到DeferredResult，最后，DeferredResult会重新申请容器线程，并将本次请求返回给客户端。
 */

@RestController
public class HystrixtestController {

    @Autowired
    private HttpService  httpService;

    private static BlockingQueue<String> blockingQueue=new ArrayBlockingQueue<>(1);

    @GetMapping("/threadtest")
    public DeferredResult<ResponseEntity<ResponseInfo>> getUser(){
        DeferredResult<ResponseEntity<ResponseInfo>> deferredResult=new DeferredResult<>(Duration.ofMinutes(5L).toMillis());
        httpService.getUser().subscribe(
                responseInfo -> deferredResult.setResult(new ResponseEntity<>(responseInfo, HttpStatus.valueOf(responseInfo.getHttpStatus().value()))),
                deferredResult::setErrorResult
        );
        return deferredResult;

    }

    /**
     * 必须消费掉，由于是异步非阻塞，所以需要睡眠3秒再返回，不然返回为empty,这也就是用DeferredResult的原因
     */
    @GetMapping("/threadtest01")
    public ResponseInfo<User> getUser01() throws InterruptedException {
        AtomicReference<ResponseInfo> user=null;
        httpService.getUser().subscribe(responseInfo -> user.set(responseInfo));
        Thread.sleep(3000);
        return user.get();
    }

    /**
     *通过设置httpService.getUser().subscribeOn(Schedulers.io())和httpService.getUser01().subscribeOn(Schedulers.io())
     * 使得httpClient在两个线程中并发请求，完成异步调用下游的目的
     *
     * httpService getUser() http-nio-8883-exec-1
     * httpService getUser01() http-nio-8883-exec-1
     * HttpClient getUser() RxIoScheduler-2
     * HttpClient getUser01() RxIoScheduler-3
     */
    @GetMapping("/threadtest02")
    public DeferredResult<ResponseEntity<List<ResponseInfo>>> getUser02() {
        DeferredResult<ResponseEntity<List<ResponseInfo>>> deferredResult = new DeferredResult(Duration.ofMinutes(5L).toMillis());
        Observable.zip(
                httpService.getUser().subscribeOn(Schedulers.io()),
                httpService.getUser01().subscribeOn(Schedulers.io()),
                (item1, item2) -> Arrays.asList(item1, item2)
        ).single().subscribe(
                item -> deferredResult.setResult(new ResponseEntity(item, HttpStatus.OK)),
                deferredResult::setErrorResult
        );
        return deferredResult;
    }


    /***
     * async 测试
     * @param httpServletResponse
     * @return
     * @throws InterruptedException
     */
    @GetMapping("/asynctest01")
    public String getAsyncTest(HttpServletResponse httpServletResponse) throws InterruptedException{
        httpServletResponse.setHeader("zhang","ssss"); //返回时随意设置一个header， 在postman中就能看到这个header https://www.cnblogs.com/c431/p/16815242.html,这个解释了 content-type
        System.out.println("getAsyncTest: "+ Thread.currentThread().getName());
        return httpService.asyncTest();
    }

    /***
     * 测试用blockQueue做限流
     * @param httpServletResponse
     * @return
     * @throws InterruptedException
     */
    @GetMapping("/asynctest02")
    public String getAsyncTest02(HttpServletResponse httpServletResponse) throws InterruptedException{
        blockingQueue.add(Thread.currentThread().getName());
        return blockingQueue.remove();

    }

    @GetMapping("/scheduled01")
    public Map getScheduledTime(){
        return httpService.scheduledBuild();
    }


}
